package com.unipay.oss;

import lombok.NonNull;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Project: yitong_parent
 * @Author: Chenmy
 * @Time: 2023-11-28 19:03:38
 * @Desc: TODO
 */

public class KafkaSenderUtils {
    private final KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 构造器方式注入 kafkaTemplate
     */
    public KafkaSenderUtils(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void send(String topicName, String msg) {
        try {

            ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topicName, msg);
            listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    //log.info("发送成功回调：{}", JSONUtil.toJsonStr(result.getProducerRecord().value()));
                    System.out.println(result.getProducerRecord().value());
                }

                @Override
                public void onFailure(@NonNull Throwable ex) {
                    System.out.println(">>>>失败原因：{}" + ex.getMessage());
                    System.out.println("发送失败回调");
                }
            });
        } catch (Exception e) {
            System.out.println("发送异常: " + e.getMessage());
        }

    }

}
